package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import com.sun.javafx.binding.StringFormatter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //绑定一个主题
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //聚合流式处理
        stream.map(((key, value) -> {
                    //重置key和value key:123  value:likes:1
                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                    return new KeyValue<>(mess.getArticleId().toString(),
                            mess.getType().toString() + ":" + mess.getAdd());
                }))
                ////按照文章id进行聚合
                .groupBy((key, value) -> key)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //自行完成聚合
                .aggregate(
                        //初始化聚合数据
                        new Initializer<String>() {
                            @Override
                            public String apply() {
                                return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                            }
                        },
                        //累加
                        new Aggregator<String, String, String>() {
                            @Override
                            public String apply(String key, String value, String aggregate) {
                                //获取上一次聚合结果
                                //准备变量
                                int col = 0, com = 0, lik = 0, vie = 0;
                                String[] aggArr = aggregate.split(",");
                                for (String agg : aggArr) {
                                    String[] split = agg.split(":");
                                    switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                                        case COLLECTION:
                                            col = Integer.parseInt(split[1]);
                                            break;
                                        case COMMENT:
                                            com = Integer.parseInt(split[1]);
                                            break;
                                        case LIKES:
                                            lik = Integer.parseInt(split[1]);
                                            break;
                                        case VIEWS:
                                            vie = Integer.parseInt(split[1]);
                                            break;
                                        default:
                                            break;
                                    }

                                    //累加本次结果
                                    String[] valAry = value.split(":");
                                    switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])) {
                                        case COLLECTION:
                                            col += Integer.parseInt(valAry[1]);
                                            break;
                                        case COMMENT:
                                            com = Integer.parseInt(valAry[1]);
                                            break;
                                        case LIKES:
                                            lik = Integer.parseInt(valAry[1]);
                                            break;
                                        case VIEWS:
                                            vie = Integer.parseInt(valAry[1]);
                                            break;
                                        default:
                                            break;
                                    }
                                }
                                String formatStr = String
                                        .format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                                System.out.println("文章的id:" + key);
                                System.out.println("当前窗口的处理结果:" + formatStr);
                                return formatStr;
                            }
                        },
                        Materialized.as("hot-atricle-stream-count-001"))
                .toStream()
                .map((key, value) -> {
                    return new KeyValue<>(key.key().toString(),formatObj(key.key().toString(),value.toString()));
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        return stream;
    }

    /**
     * 格式化消息的value数据
     */
    private String formatObj(String key, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(key));

        String[] arr = value.split(",");
        for (String typeAndAdd : arr) {
            String[] split = typeAndAdd.split(":");

            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
                default:
                    break;
            }
        }
        return JSON.toJSONString(mess);
    }
}